/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月20日 下午10:01:42
*/
package cn.com.guage.flink.window;

/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月20日 下午10:01:42
*/
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

//每10秒滚动统计一次且不叠加
public class TumblingindowProcessTest {
	public static void main(String[] args) {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStream<String> ds = env.socketTextStream("192.168.0.128", 9528);

		ds.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
			/**
			 * 
			 */
			private static final long serialVersionUID = 4098445035431416797L;

			@Override
			public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) {
				for (String str : s.split(",")) {
					collector.collect(Tuple2.of(str, 1));
				}

			}
		}).keyBy(new KeySelector<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, Integer> getKey(Tuple2<String, Integer> value) throws Exception {
				return Tuple2.of(value.f0, value.f1);
			}

		}).window(TumblingProcessingTimeWindows.of(Time.seconds(10))).sum(1).print();
		try {
			env.execute();
		} catch (Exception e) {
			e.printStackTrace();
		}

	}
}